package com.example.demo.hutool.redis.cluster;


import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.*;
import redis.clients.jedis.util.JedisClusterCRC16;

import java.util.*;

public class JedisClusterPipeline extends JedisCluster {

    public JedisClusterPipeline(Set<HostAndPort> nodes, GenericObjectPoolConfig poolConfig) throws Exception {
        super(nodes, poolConfig);
        super.connectionHandler = new JedisSlotAdvancedConnectionHandler(nodes, poolConfig, 200);
    }

    public JedisSlotAdvancedConnectionHandler getConnectionHandler() {
        return (JedisSlotAdvancedConnectionHandler) this.connectionHandler;
    }

    /**
     * 刷新集群信息，当集群信息发生变更时调用
     *
     * @param
     * @return
     */
    public void refreshCluster() {
        connectionHandler.renewSlotCache();
    }

    public static void main(String[] args) throws Exception {
        JedisPubSub jedisPubSub = new JedisPubSub() {
            @Override
            public void onMessage(String channel, String message) {
                System.out.println("onMessage" + "---" + channel + ":" + message);
                try {
                    Thread.sleep(9000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                if (message.equals("kkk")) {
                    throw new RuntimeException("myException");
                }
            }

            @Override
            public void onPUnsubscribe(String pattern, int subscribedChannels) {
                System.out.println("onPUnsubscribe" + "---" + pattern + ":" + subscribedChannels);
            }

            @Override
            public void onPMessage(String pattern, String channel, String message) {
                System.out.println("onPMessage" + "---" + pattern + "-----" + channel + "-----" + message);
            }

            @Override
            public void onSubscribe(String channel, int subscribedChannels) {
                System.out.println("onSubscribe" + "---" + channel + ":" + subscribedChannels);
            }
        };
        JedisClusterPipeline jedisClusterPipeline = getJedisClusterPipeline();
        ArrayList<String> arrayList = new ArrayList<>();
        arrayList.add("sxm");
        arrayList.add("sxm1");
        arrayList.add("sxm2");
        arrayList.add("sxm3");
        jedisClusterPipeline.refreshCluster();
        Map<String, JedisPool> nodes = getStringJedisPoolMap(jedisClusterPipeline);

        System.out.println(nodes.toString());


        extracted(jedisPubSub, nodes);
        hgetPipeline(jedisClusterPipeline, null);
    }

    public static JedisClusterPipeline getJedisClusterPipeline() throws Exception {
        HashSet<HostAndPort> hostAndPorts = new HashSet<>();
        hostAndPorts.add(new HostAndPort("172.20.54.122", 7291));
        hostAndPorts.add(new HostAndPort("172.20.54.122", 7292));
        hostAndPorts.add(new HostAndPort("172.20.54.122", 7293));
        hostAndPorts.add(new HostAndPort("172.20.54.122", 7294));
        hostAndPorts.add(new HostAndPort("172.20.54.122", 7295));
        hostAndPorts.add(new HostAndPort("172.20.54.122", 7296));
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        // 最大连接数
        jedisPoolConfig.setMaxTotal(10);
        // 最大连接阻塞等待时间
        jedisPoolConfig.setMaxWaitMillis(-1);
        // 最大空闲连接数
        jedisPoolConfig.setMaxIdle(1);
        // 最小空闲连接数
        jedisPoolConfig.setMinIdle(1);
        jedisPoolConfig.setTestOnBorrow(true);
        jedisPoolConfig.setTestOnReturn(true);
        JedisClusterPipeline jedisClusterPipeline = new JedisClusterPipeline(hostAndPorts, jedisPoolConfig);
        return jedisClusterPipeline;
    }

    private static void extracted(JedisPubSub jedisPubSub, Map<String, JedisPool> nodes) {
        HashSet<String> set = new HashSet<>();
        for (Map.Entry<String, JedisPool> entry : nodes.entrySet()) {
            set.add(entry.getKey());
            new Thread(() -> {
                Map.Entry<String, JedisPool> stringJedisPoolEntry = entry;
                String key = stringJedisPoolEntry.getKey();
                while (true) {
                    try {
                        JedisPool jedisPool = stringJedisPoolEntry.getValue();
                        Jedis resource = jedisPool.getResource();
                        Set<String> keys = resource.keys("*");
                        System.out.println("node:--" + key + "---dbsize:" + keys.size());
                        List<String> strings = resource.configGet("notify-keyspace-events");

                        String keyspace = strings.get(1);
                        if (keyspace == null || "".equals(keyspace) || !keyspace.contains("A") || !keyspace.contains("K") && !keyspace.contains("E")) {
                            resource.configSet("notify-keyspace-events", "AKE");
                        }
                        resource.psubscribe(jedisPubSub, "__keyspace@0__:*");
                    } catch (Exception e) {
                        if (stringJedisPoolEntry != null) {
                            set.remove(stringJedisPoolEntry.getKey());
                            stringJedisPoolEntry = null;
                        }
                        try {
                            Thread.sleep(1000);
                            JedisClusterPipeline jedisClusterPipeline = getJedisClusterPipeline();
                            Map<String, JedisPool> stringJedisPoolMap = getStringJedisPoolMap(jedisClusterPipeline);
                            for (Map.Entry<String, JedisPool> jedisPoolEntry : stringJedisPoolMap.entrySet()) {
                                if (!set.contains(jedisPoolEntry.getKey())){
                                    stringJedisPoolEntry=jedisPoolEntry;
                                    break;
                                }
                            }
                        } catch (Exception ex) {
                            continue;
                        }
                    }
                }
            }).start();
        }
    }

    private static Map<String, JedisPool> getStringJedisPoolMap(JedisClusterPipeline jedisClusterPipeline) {

        jedisClusterPipeline.refreshCluster();
        Map<String, JedisPool> nodes = jedisClusterPipeline.getConnectionHandler().getNodes();
        Iterator<Map.Entry<String, JedisPool>> iterator = nodes.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, JedisPool> next = iterator.next();
            JedisPool jedisPool = next.getValue();
            Jedis resource = jedisPool.getResource();
            String replication = resource.info("Replication");
            String[] split = replication.split("\r\n");
            boolean isDelete = true;
            for (String info : split) {
                String[] key = info.split(":");
                if (key[0].equals("role")) {
                    if (key[1].equals("master")) {
                        isDelete = false;
                        break;
                    }
                }
            }
            if (isDelete) {
                iterator.remove();
            }
        }
        return nodes;
    }

    public static List<byte[]> hgetPipeline(JedisClusterPipeline jedisClusterPipeline, List<String> keyFields) {
        List<byte[]> result = new ArrayList<>();
        Map<JedisPool, List<String>> poolKeys = new HashMap<>();
        JedisSlotAdvancedConnectionHandler jedisSlotAdvancedConnectionHandler = jedisClusterPipeline.getConnectionHandler();
        jedisClusterPipeline.refreshCluster();
//        System.out.println(jedisClusterPipeline.get("3"));
        int slot = JedisClusterCRC16.getSlot("3".getBytes());
        JedisPool jedisPool = jedisSlotAdvancedConnectionHandler.getJedisPoolFromSlot(slot);
        Jedis resource = jedisPool.getResource();
        System.out.println(resource.getClient().getSocket().getPort());
        System.out.println(resource.getClient().getSocket().getLocalPort());
        System.out.println(resource.get("3"));
        resource.close();
        return null;
    }

}
